package org.future.task.core.handler;

import org.future.task.core.cache.ParamCache;
import org.future.task.core.monitor.Monitor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public abstract class RunnableTask<Flow> implements TaskRunner<Flow> {

    private static final AtomicInteger ato = new AtomicInteger(1);

    private String queueName;

    private TransForm<Flow> transForm;

    private Monitor monitor;

    private int timeOut = -1;

    private boolean runStatus = true;

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public TransForm<Flow> getTransForm() {
        return transForm;
    }

    public void setTransForm(TransForm<Flow> transForm) {
        this.transForm = transForm;
    }

    public int getTimeOut() {
        return timeOut;
    }

    public void setTimeOut(int timeOut) {
        this.timeOut = timeOut;
    }

    public Monitor getMonitor() {
        return monitor;
    }

    public void setMonitor(Monitor monitor) {
        this.monitor = monitor;
    }

    @Override
    public Flow transform(Object o) {
        return (Flow) o;
    }

    public void runnable(int batchSize) {
        Thread thread = Thread.currentThread();
        thread.setName("task-" + this.queueName + "-" + ato.getAndIncrement());
        ParamCache.setTaskThread(this.queueName, thread);

        Connection connection = ParamCache.getConnection(this.queueName);

        String threadName = Thread.currentThread().getName();

        String className = this.getClass().getName();

        while (this.runStatus) {
            try {
                List<Object> dataList = connection.getDataList(this.queueName, batchSize <= 0 ? 300 : batchSize, timeOut);
                if (dataList == null) {
                    continue;
                }
                dataList.removeAll(Collections.singleton(null));
                if (dataList.isEmpty()) {
                    continue;
                }
                List<Flow> before = before(dataList);
                runner(before);
                after(before);
                if (monitor != null) {
                    monitor.inc(className, dataList.size(), queueName, threadName);
                }
            } catch (Exception e) {
                try {
                    tryException(null, e);
                } catch (Exception e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    @Override
    public void runner(List<Flow> runner) throws Exception {
        for (Flow flow : runner) {
            try {
                execute(flow);
            } catch (Exception e) {
                tryException(flow, e);
            }
        }
    }

    @Override
    public List<Flow> before(List<Object> object) {

        List<Flow> flows = new ArrayList<>();
        Flow transform;
        for (Object obj : object) {
            if (transForm != null) {
                transform = transForm.transform(obj);
            } else {
                transform = transform(obj);
            }
            flows.add(transform);
        }
        return flows;
    }

    @Override
    public List<Flow> after(List<Flow> flows) {
        return null;
    }

    @Override
    public void tryException(Flow flow, Exception e) {
        e.printStackTrace();
    }
}
